package edu.hda.pt.client.connections;

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;

import edu.hda.pt.client.PTClient;
import edu.hda.pt.client.connections.netty.AsyncClientHandler;
import edu.hda.pt.client.connections.netty.ClientHandler;
import edu.hda.pt.client.connections.netty.ClientPipelineFactory;
import edu.hda.pt.client.connections.netty.SyncClientHandler;
import edu.hda.pt.utils.Stream;
import edu.hda.pt.utils.TestMethod;
import edu.hda.pt.utils.TestObject;
import edu.hda.pt.utils.Stream.StreamState;
import edu.hda.pt.utils.connections.netty.NettyConstants;

/**
 * Client implementation for the connection method JBoss Netty
 * 
 * @author Marc Hammerton
 * 
 */
public class PTClientNetty extends PTClient {

	private ClientBootstrap bootstrap;
	private ChannelFuture cf;
	private Channel channel;
	private ClientHandler handler = null;

	public Stream stream = null;

	private String serverIP;

	/**
	 * Constructor for setting up the Netty client.
	 * 
	 * @param serverIP
	 *            - The IP address of the Netty server
	 */
	public PTClientNetty(String serverIP) {
		this.serverIP = serverIP;

		ChannelFactory factory = new NioClientSocketChannelFactory(Executors
				.newCachedThreadPool(), Executors.newCachedThreadPool());

		bootstrap = new ClientBootstrap(factory);
		bootstrap.setPipelineFactory(new ClientPipelineFactory(this));

	}

	/**
	 * @see edu.hda.pt.client.PTClient#connect()
	 */
	@Override
	public void connect() {
		cf = bootstrap.connect(new InetSocketAddress(this.serverIP,
				NettyConstants.PORT));

		channel = cf.getChannel();

		try {
			if (PTClient.test2run == TestMethod.STREAMING) {
				handler = (AsyncClientHandler) cf.await().getChannel()
						.getPipeline().getLast();
			} else {
				handler = (SyncClientHandler) cf.await().getChannel()
						.getPipeline().getLast();
			}
		} catch (InterruptedException e) {
			System.out.println(this.clientID + " Exception:");
			e.printStackTrace();
		}

	}

	/**
	 * @see edu.hda.pt.client.PTClient#disconnect()
	 */
	@Override
	public void disconnect() {
		channel.disconnect();

	}

	/**
	 * @see edu.hda.pt.client.PTClient#sendBool(int, boolean)
	 */
	@Override
	protected void sendBool(int messageId, boolean boolPrim) {
		channel.write(boolPrim);

		handler.responseReceived();
	}

	/**
	 * @see edu.hda.pt.client.PTClient#sendBytes(int, byte[])
	 */
	@Override
	protected void sendBytes(int messageId, byte[] bytes) {
		channel.write(bytes);

		handler.responseReceived();
	}

	/**
	 * @see edu.hda.pt.client.PTClient#sendInt(int, int)
	 */
	@Override
	protected void sendInt(int messageId, int intPrim) {
		channel.write(intPrim);

		handler.responseReceived();
	}

	/**
	 * @see edu.hda.pt.client.PTClient#sendLong(int, long)
	 */
	@Override
	protected void sendLong(int messageId, long longPrim) {
		channel.write(longPrim);

		handler.responseReceived();
	}

	/**
	 * @see edu.hda.pt.client.PTClient#sendObject(int,
	 *      edu.hda.pt.utils.TestObject)
	 */
	@Override
	protected void sendObject(int messageId, TestObject testObject) {
		channel.write(testObject);

		handler.responseReceived();
	}

	/**
	 * @see edu.hda.pt.client.PTClient#sendString(int, java.lang.String)
	 */
	@Override
	protected void sendString(int messageId, String text) {
		channel.write(text);

		handler.responseReceived();
	}

	/**
	 * @see edu.hda.pt.client.PTClient#sendStreamingRequest(int, int)
	 */
	@Override
	protected void sendStreamingRequest(int fs) {
		System.out.println(this.clientID + ": requesting stream");
		stream = new Stream(fs);
		stream.setState(StreamState.START);
		channel.write(stream);
	}

	/**
	 * @see edu.hda.pt.client.PTClient#terminateStreaming()
	 */
	@Override
	protected void terminateStreaming() {
		if (stream != null && stream.getState() == StreamState.RUNNING) {
			stream.setState(StreamState.FINISH);
			channel.write(stream);
			try {
				Thread.sleep(15000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			System.out.println(this.clientID + ": stopping stream");
		}
	}

}
